package com.ebs.md.fxmonitor.client.impl;

import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import org.apache.log4j.Logger;

import com.ebs.commons.comp.IParsedPayload;
import com.ebs.commons.comp.IPubSubChannel;
import com.ebs.commons.comp.ParsedPayload;
import com.ebs.commons.db.DataSourceException;
import com.ebs.commons.db.SPLoader;
import com.ebs.commons.logger.MessageText;
import com.ebs.commons.threads.EbsThreadFactory;
import com.ebs.md.fxmonitor.topics.ClientTopicKeys;

public class RecoverySynch implements MessageListener
{
	private String SP_FROM_TIME_TAG = "TIME_FROM";
	private String SP_TO_TIME_TAG = "TIME_TO";
	private String SP_ARB_SESSION_TAG = "ARB_SESSION_NUM";
	private String SP_ARB_MSG_NUM_TAG = "ARB_MSG_NUM";
	private String SP_MSG_LIST = "MSG_LIST";
	
	private static Logger s_logger = Logger.getLogger(RecoverySynch.class);
	
	private SPLoader m_SPLoader;
	
	private int m_FirstTimestamp = 0;
	private int m_FirstSessionNum = 0;
	private int m_FirstMsgSeqNum = 0;
	
	private int m_BeginRecoveryTimestamp = 0;
	
	private int m_current_time_from = 0;
	private int m_current_time_to = 0;
	private int m_current_session = 0;
	private int m_current_msg_num = 0;
	
	private int m_recovery_interval = 10;
	private int m_recovery_delay = 180;
	private int m_recovery_ratio = 2;
	
	private int m_periodic_request_interval = 1;
	
	private IPubSubChannel m_NdsChannel;
	
	private String m_publishToTopicName = "fxm_messages";
	
	//0=Disconnected 1=StartRecovery 2=RecoveryInProgress 3=Connected
	private int m_currentStatus;
	
	private ArbMsgSP m_msgToSpMapping;
	
	private ScheduledExecutorService m_scheduledExecutor;
	private int m_dealStoragePeriodSec;
	
	private List<Integer> m_msgTypeToSubscribe;
	
	private Map<String, List<Integer>> m_requiredMsgTypes;
    private Map<String,List<Integer>> m_spToMsgTypeMap;
    private List<String> m_spNames;
    
    private boolean m_obRecoveryPresent = false;
    
    private long m_SessionNumber;
    
    private Object m_Monitor = new Object();
    
    private ArrayBlockingQueue<IParsedPayload> m_synchQueue = new ArrayBlockingQueue<IParsedPayload>(500000);
    
    PeriodicRecovery m_PeriodicRecovery = new PeriodicRecovery();
    
    
    public RecoverySynch (IPubSubChannel ndsChannel)
    {
    	m_NdsChannel = ndsChannel;
    }
    	
    
    public void init()
    {
    	logout();
    	m_spToMsgTypeMap = m_msgToSpMapping.getSpPerSubscription(m_msgTypeToSubscribe);
    	orderSpList();
    	m_scheduledExecutor = Executors.newSingleThreadScheduledExecutor(new EbsThreadFactory("recoverySync periodicRecovery"));
    	s_logger.info("RecoverySync init() is done");
    }
    
    public void setSpLoader( SPLoader spLoader)
    {
    	m_SPLoader = spLoader;
    }
    
    public void setMsgToSpMapping(ArbMsgSP msgToSpMap)
    {
    	m_msgToSpMapping = msgToSpMap;
    }
	
    public void setRecoveryInterval(int recoveryIntervalSec)
    {
    	m_recovery_interval = recoveryIntervalSec;
    }
    
    public void setRecoveryDelay(int recoveryDelaySec)
    {
    	m_recovery_delay = recoveryDelaySec;
    }
    
    public void setRecoveryRatio(int recoveryRatio)
    {
    	m_recovery_ratio = recoveryRatio;
    }
        
    public void setSubscriptionList(Map<Integer,String> subscribeToMessages)
    {   
    	m_msgTypeToSubscribe = new ArrayList<Integer>();
    	m_requiredMsgTypes = new HashMap<String,List<Integer>>();
    	
    	if(subscribeToMessages == null)
    	{
    		return;
    	}
    	
    	for(Integer nextMsgType : subscribeToMessages.keySet())
    	{
    		m_msgTypeToSubscribe.add(nextMsgType);
    		String mq_msg_type = subscribeToMessages.get(nextMsgType);
    		if(m_requiredMsgTypes.containsKey(mq_msg_type))
    		{
    			m_requiredMsgTypes.get(mq_msg_type).add(nextMsgType);
    		}
    		else
    		{
    			List<Integer> nextMsgTypes = new ArrayList<Integer>();
    			nextMsgTypes.add(nextMsgType);
    			m_requiredMsgTypes.put(mq_msg_type, nextMsgTypes);
    		}
    	}
    	
    	
    }
    
    public void setPublishMsqToTopic(String topicName)
    {
    	m_publishToTopicName = topicName;
    }
    
    public void orderSpList()
    {
    	m_spNames = new ArrayList<String>();
    	for(String nextSpName : m_spToMsgTypeMap.keySet())
    	{
    		if(nextSpName.equalsIgnoreCase("GET_HIST_OB"))
    		{
    			m_spNames.add(nextSpName);
    		}
    	}
    	
    	for(String nextSpName : m_spToMsgTypeMap.keySet())
    	{
    		if(!nextSpName.equalsIgnoreCase("GET_HIST_OB"))
    		{
    			m_spNames.add(nextSpName);
    		}
    	}
    }
    
	private synchronized void submitTheBuffer()
	{
		if(m_synchQueue.size() > 0)
		{
			s_logger.info("RecoverySync m_synchQueue.size() is:" + m_synchQueue.size());
			IParsedPayload payload;
			while ((payload = m_synchQueue.poll()) != null)
			{		
				if((m_SessionNumber == 0)||(m_currentStatus == 0))
				{
					//drop to the floor, session was interrupted
//					s_logger.info("RecoverySync Drop to the floor. SessionNumber or currentStatus are 0 ie session was interrupted. ");
				}
				else
				{
					//send msg to ClientMediator's topic
					s_logger.info("RecoverySync submitTheBuffer payload:" + payload.getParsedMessage());
					this.submitInternalMessage(m_publishToTopicName, payload);
				}
			}			
		}
		else
		{
			s_logger.info("RecoverySync m_synchQueue.size() is 0");
		}
		m_currentStatus = 3;
		s_logger.info("RecoverySync submitTheBuffer is done. m_currentStatus=3 REAL_TIME");
	}

    
    /**
     * From MessageListener
     * @param message Message - message from the router
     */
    public void onMessage(Message message) 
    {
        try
        {
            IParsedPayload payload = (IParsedPayload) ((ObjectMessage) message).
                                     getObject();
            int msg_type = payload.getParsedIntField("type");            
            String str_msg_type = "" + msg_type;
            String mq_msg_type = payload.getParsedStringField("mq_msg_type","");
//            if(!mq_msg_type.equalsIgnoreCase("RTCK"))
//            {
//            	s_logger.info("RecoverySync m_SessionNumber:" + m_SessionNumber + 
//            		" m_currentStatus:" + m_currentStatus + " has received message:" + payload.getParsedMessage());
//            }
            if(str_msg_type.equalsIgnoreCase("-4"))
            {
            	//we've received unsubscribe-subscribe message
                int discriminator = payload.getParsedIntField(ClientTopicKeys.SubscribeType);
                if (discriminator > 0)
                {
                    onSubscribeRequest(payload);
                }
                else
                {
                    onUnsubscribeRequest();
                }
                return;
            }
            
            if(mq_msg_type.length() > 0)
            {            	
            	//we've received RDS message
                if((m_SessionNumber == 0)||(m_currentStatus == 0))
                {
                	//we were reset, not connected drop data to the floor
                	//s_logger.info("RecoverySync drop data to the floor. m_SessionNumber:" + m_SessionNumber +
                	//		" m_currentStatus:" + m_currentStatus);
                	return;
                }
                else if(m_currentStatus == 1)
                {
                	//we are in start recovery mode. wait for the first tick to open access to the queue                	
                	if(!mq_msg_type.equalsIgnoreCase("RTCK"))
                	{
//                		s_logger.info("RecoverySync m_currentStatus == 1 msg is not TICK drop to the floor msg:" + payload.getParsedMessage());
                		return;
                	}
                	else                		
                	{
                		synchronized(m_Monitor)
                    	{
                			int currentTime = ((Long)(System.currentTimeMillis()/1000)).intValue();
                    	
                			if(((currentTime - m_current_time_to)/m_recovery_interval) <= m_recovery_ratio)
                			{
                				m_FirstTimestamp = payload.getParsedIntField("arb_msg_time",0);
                				m_FirstSessionNum = payload.getParsedIntField("session_number",0);
                				m_FirstMsgSeqNum = payload.getParsedIntField("msg_seq_num",0);
                				m_currentStatus = 2;
                				s_logger.info("RecoverySync TICK m_currentStatus is moved to 2");                    		
                			}
//                			else
//                			{
//                				s_logger.info("RecoverySync m_currentStatus == 1 msg TICK. m_current_time_to more than m_recovery_ratio. Continue recovery. No sync currentTime:" + currentTime +
//                						" m_current_time_to:" + m_current_time_to + 
//                						" m_recovery_interval:" + m_recovery_interval + 
//                						" ratio:" + ((currentTime - m_current_time_to)/m_recovery_interval));
//                			}
                    	}
                	}
                }
                else if(m_currentStatus == 2)
                {
                	//we are accumulating the data into the queue.                 	
                	if(!m_requiredMsgTypes.containsKey(mq_msg_type))
                	{
                		if(!mq_msg_type.equalsIgnoreCase("RTCK"))
                		{
                			s_logger.info("RecoverySync state is ACCUMULATE Message is not in required list. Drop to the floor msg:" + payload.getParsedMessage() );
                		}
                		return;
                	}
                    if(!mq_msg_type.equalsIgnoreCase("RTCK"))
                    {
                    	s_logger.info("RecoverySync m_SessionNumber:" + m_SessionNumber + 
                    		" m_currentStatus:" + m_currentStatus + " has accumulated message:" + payload.getParsedMessage());
                    }

                	m_synchQueue.offer(payload);
                }
                else if(m_currentStatus == 3)
                {
                	if(mq_msg_type.equalsIgnoreCase("EVNT"))
                	{
                		//this is an event. If this is change of bin files, force disconnect?
                	}
                	// we are in real time. 
                	if(!m_requiredMsgTypes.containsKey(mq_msg_type))
                	{
                		if(!mq_msg_type.equalsIgnoreCase("RTCK"))
                		{
                			s_logger.info("RecoverySync state is REAL TIME. Message is not in requireed list. Drop to the floor msg:" + payload.getParsedMessage() );
                		}
                		return;
                	}
                	//Send msg to ClientMediator's topic
                	this.submitInternalMessage(m_publishToTopicName, payload);                	
                }            	
            }
 
        }
        catch(JMSException jmse)
        {
            s_logger.error(new MessageText("NETLINK_0100004", message), jmse);
            onUnsubscribeRequest();
            /** @todo Disconnect all */
        }    	
    }
    
    
    public void onUnsubscribeRequest()
    {
    	resetValues();
    }
    
	private void startRecoveryThread()
	{    	
    	m_scheduledExecutor.schedule(m_PeriodicRecovery, m_periodic_request_interval, TimeUnit.SECONDS);    	
	}


    public void onSubscribeRequest(IParsedPayload payload)
    {
    	resetValues();
    	m_SessionNumber = payload.getParsedIntField(ClientTopicKeys.SubscriptionId, 0);
    	//begin recovery 11 min before to get OB snapshots
    	m_BeginRecoveryTimestamp = ((Long)(System.currentTimeMillis()/1000)).intValue() - m_recovery_delay;
    	m_current_time_from = m_BeginRecoveryTimestamp;
    	// set 10 sec(or config) intervals for recovery
    	m_current_time_to = m_current_time_from + m_recovery_interval;
    	m_currentStatus = 1;
    	startRecoveryThread();
    }
    
     
    public void logout()
    {
    	m_SessionNumber = 0;
    	resetValues();
    }
    
    public void resetValues()
    {
    	m_FirstTimestamp = 0;
    	m_FirstSessionNum = 0;
		m_FirstMsgSeqNum = 0;
		m_BeginRecoveryTimestamp = 0;
		m_currentStatus = 0;
		m_current_time_from = 0;
		m_current_time_to = 0;
		m_current_session = 0;
		m_current_msg_num = 0;	
		m_obRecoveryPresent = false;
		m_periodic_request_interval = 1;
		if(m_synchQueue != null)
		{
			m_synchQueue.clear();
		}
		m_synchQueue = new ArrayBlockingQueue<IParsedPayload>(500000);
    }
    
	
	private void submitInternalMessage(String topicName, IParsedPayload payload)
	{
		try
		{			
			m_NdsChannel.submitMessage(topicName, payload);
		}
		catch(JMSException jmse)
		{
			s_logger.error("RecoverySynchronizer JMSException:",jmse);
		}
	}	
    
    public void requestDBRecoveryBatch()
    {
    	s_logger.info("recoverySync Begin periodic requestDBRecoveryBatch()");
    	if(m_currentStatus == 3)
    	{
    		// REAL_TIME we are done with recovery do not start timer thread
    		s_logger.info("SubscriptionPaginator m_currentStatus is REAL_TIME. End of Recovery");
    		return;
    	}
    	
    	if(m_currentStatus == 0)
    	{
    		// Status disconnected the recovery was cancelled
    		s_logger.info("SubscriptionPaginator m_currentStatus is SILENT. Recovery interrupted");
    		return;    		
    	}    
    	
    	int currentTime = ((Long)(System.currentTimeMillis()/1000)).intValue();
    	
    	TreeMap<String,IParsedPayload> recoveryResult = new TreeMap<String,IParsedPayload>();
    	
    	Map<String,Object> params = new HashMap<String, Object>();
    	
    	params.put(SP_FROM_TIME_TAG, m_current_time_from);
    	params.put(SP_TO_TIME_TAG, m_current_time_to);
    	params.put("CURRENT_TIME", currentTime);
    	
    	if(m_current_session > 0)
    	{
    		params.put(SP_ARB_SESSION_TAG, m_current_session);
    	}
    	if(m_current_msg_num > 0)
    	{
    		params.put(SP_ARB_MSG_NUM_TAG, m_current_msg_num);
    	}
    	
    	accumulateRecoveryResult(recoveryResult, params);
    	int internal_status = publishAccumulatedResults(recoveryResult);
    	if(internal_status == 0)
    	{
    		//we continue recovery
    		startRecoveryThread();  
    		return;
    	}
    	if(internal_status == 2)
    	{
    		s_logger.info("RecoverySync during RequestDB internal status was changed to remove Recovery mode.");
    	}
    	else if(internal_status == 1)
    	{
    		//recovery is over change status to real time and send queue
    		submitTheBuffer();
    	}    	
    }
    
    
    
    
    private void accumulateRecoveryResult(TreeMap<String,IParsedPayload> accumulator, Map<String,Object> params)
    {
    	if(m_currentStatus == 0)
    	{
    		s_logger.info("RecoverySync State is SILENT. Recovery dropped.");
    		accumulator = new TreeMap<String,IParsedPayload>();
    		return;
    	}
    	
    	boolean obPresent = false;
    	
    	int totalSize = 0;
    	
    	for(int j=0; j < m_spNames.size(); j++)
    	{
    		String nextSpName = m_spNames.get(j);
	    	if(m_currentStatus == 0)
	    	{    		    		
	    		s_logger.info("RecoverySync Accumulation stopped. State is SILENT. Recovery dropped.");
	    		accumulator = new TreeMap<String,IParsedPayload>();
	    		return;
	    	}
    		params.put(SP_MSG_LIST, m_spToMsgTypeMap.get(nextSpName));
    		params.put(m_SPLoader.getSPLabel(), nextSpName);
    		try
    		{
        		s_logger.info("RecoverySync DBRecovery params:" + params);
    			List<Map<String,Object>> result = m_SPLoader.getRecordSet(params);    			
    			totalSize = totalSize + result.size();
    			s_logger.info("RecoverySync DBRecovery result size:" + result.size());
    			if(m_currentStatus == 0)
		    	{    		    		
		    		s_logger.info("RecoverySync Accumulation stopped. State is SILENT. Recovery dropped.");
		    		accumulator = new TreeMap<String,IParsedPayload>();
		    		return;
		    	}
    			if((nextSpName.equalsIgnoreCase("GET_HIST_OB")) && (!m_obRecoveryPresent) && (totalSize < 1))
    			{
    				s_logger.info("recoverySync GET_HIST_OB return size is 0. move to the next interval");
    				return;
    			}
    			else if((nextSpName.equalsIgnoreCase("GET_HIST_OB")) && (!m_obRecoveryPresent) && (totalSize > 0))
    			{
    				m_obRecoveryPresent = true;
    			}
    			
    			for(int i = 0; i < result.size(); i++)
        		{
    		    	if(m_currentStatus == 0)
    		    	{    		    		
    		    		s_logger.info("RecoverySync Accumulation stopped. State is SILENT. Recovery dropped.");
    		    		accumulator = new TreeMap<String,IParsedPayload>();
    		    		return;
    		    	}
    				Map<String,Object> nextRecord = result.get(i);
        			IParsedPayload nextPayload = new ParsedPayload(); 
        			String recordOrderKey = "";
        			for(String nextKey : nextRecord.keySet())
        			{
        				Object nextValue = nextRecord.get(nextKey);
        				if(nextValue != null)
        				{
        					nextPayload.addField(nextKey, nextValue);
        				}
        			}
   			
        			nextPayload.addField(ClientTopicKeys.SubscriptionId, m_SessionNumber);
        			nextPayload.addField("recovery_current_flag", "0");
        			
        			NumberFormat formatter = new DecimalFormat("0000000000");   
        			recordOrderKey = recordOrderKey + formatter.format(nextPayload.getParsedIntField(ClientTopicKeys.ArbMessageTime,0));
        			recordOrderKey = recordOrderKey + formatter.format(nextPayload.getParsedIntField(ClientTopicKeys.ArbSessionNum,0));
        			recordOrderKey = recordOrderKey + formatter.format(nextPayload.getParsedIntField(ClientTopicKeys.ArbSequenceNum,0));
        			recordOrderKey = recordOrderKey + formatter.format(nextPayload.getParsedIntField("component_number",0));
        			recordOrderKey = recordOrderKey + nextPayload.getParsedStringField("mq_msg_type","ZZZZ");
        			accumulator.put(recordOrderKey, nextPayload);        			
        		}
    		}
    		catch(DataSourceException dse)
    		{
    			dse.printStackTrace();
        		s_logger.error("SubscriptionPaginator Exception dse:", dse);
//        		sendRdbError();
    		}
    	}
		s_logger.info("RecoverySync Got result from DB size:" + totalSize + 
				" recovery interval:" + m_recovery_interval);		
		
    }

    
    public int publishAccumulatedResults(TreeMap<String,IParsedPayload> recoveryResult)    
    {
    	int toReturn = 0;
    	for(IParsedPayload nextPayload : recoveryResult.values())
    	{
    		if((m_currentStatus < 1)||(m_currentStatus > 2))
    		{
    			// we are not in recovery anymore. drop msg to the floor
    			toReturn = 2;
    			return toReturn;
    		}
    		else
    		{	
    			// we are in recovery. check whether we've reached the real time
    			if(isRecoveryOver(nextPayload))
    			{
    				//the Real Time criteria reached
    				toReturn = 1;
        			return toReturn;
    				
    			}
    			else
    			{
    				submitInternalMessage(m_publishToTopicName,nextPayload); 
    			}
    		}
    	}
    	
    	synchronized(m_Monitor)
    	{
    		m_current_time_from = m_current_time_to;    	
    		m_current_time_to = m_current_time_from + m_recovery_interval;
    	
    		int currentTime = ((Long)(System.currentTimeMillis()/1000)).intValue();
    	
    		if((m_current_time_from >= currentTime)&&(m_FirstTimestamp <= 0))
    		{
    			//Start time more than current time and TICK has not arrived. Move to REAL TIME and wait for messages
    			s_logger.info("RecoverySync Recovery from m_current_time_from:" + m_current_time_from + 
    				" currentTime:" + currentTime +
    				" TICK has not arrived. Moving to REAL TIME");
    			toReturn = 1;
    			return toReturn;
    		}
    		else if(m_current_time_from >= currentTime)
    		{
    			//we are past current time. Stop recovery
    			s_logger.info("RecoverySync Recovery from m_current_time_from:" + m_current_time_from + 
    				" currentTime:" + currentTime +
    				" Moving to REAL TIME");
    			toReturn = 1;
    			return toReturn;
    		}
    	
//    		s_logger.info("RecoverySynch scheduled next run " +
//    			"m_current_time_from:" + m_current_time_from + 
//    			" m_current_time_to:" + m_current_time_to +
//    			" currentTime:" + currentTime);    	    		
    	}
    	return toReturn;
  }
    
    

	private boolean isRecoveryOver(IParsedPayload nextPayload)
	{
		if((m_currentStatus < 1)||(m_currentStatus > 2))
		{
			//		s_logger.info("RecoverySync current status is not SYNCH isRecoveryOver return false");
			return false;
		}
	
		if(m_FirstTimestamp <= 0)
		{
			//		s_logger.info("RecoverySync first TICK has not arrived yet m_FirstTimestamp=" + m_FirstTimestamp + " isRecoveryOver return false");
			return false;
		}
	
		int msg_arb_timestamp = nextPayload.getParsedIntField(ClientTopicKeys.ArbMessageTime);
		if((m_FirstTimestamp > 0)&& (msg_arb_timestamp > m_FirstTimestamp))
		{
			s_logger.info("SubscriptionPaginator m_FirstTimestamp=" + m_FirstTimestamp +
				" msg_arb_timestamp=" + msg_arb_timestamp +
				" isRecoveryOver returns true");
			return true;
		}
	
		int msg_arb_session = nextPayload.getParsedIntField(ClientTopicKeys.ArbSessionNum);
		if((msg_arb_timestamp >= m_FirstTimestamp)&&(msg_arb_session > m_FirstSessionNum))
		{
			s_logger.info("SubscriptionPaginator m_FirstTimestamp=" + m_FirstTimestamp +
				" msg_arb_timestamp=" + msg_arb_timestamp +
				" m_FirstSessionNum=" + m_FirstSessionNum +
				" msg_arb_session=" + msg_arb_session +
				" isRecoveryOver returns true");    		
			return true;
		}
	
		int msg_arb_seq = nextPayload.getParsedIntField(ClientTopicKeys.ArbSequenceNum);
		if((msg_arb_timestamp >= m_FirstTimestamp)
			&&(msg_arb_session >= m_FirstSessionNum)
			&&(msg_arb_seq >= m_FirstMsgSeqNum))
		{
			s_logger.info("SubscriptionPaginator m_FirstTimestamp=" + m_FirstTimestamp +
				" msg_arb_timestamp=" + msg_arb_timestamp +
				" m_FirstSessionNum=" + m_FirstSessionNum +
				" msg_arb_session=" + msg_arb_session +
				" m_FirstMsgSeqNum=" + m_FirstMsgSeqNum +
				" msg_arb_seq=" + msg_arb_seq +
				" isRecoveryOver returns true");    		    		
			return true;
		}	 	 
		//	s_logger.info("SubscriptionPaginator m_FirstTimestamp=" + m_FirstTimestamp +
		//			" msg_arb_timestamp=" + msg_arb_timestamp +
		//			" m_FirstSessionNum=" + m_FirstSessionNum +
		//			" msg_arb_session=" + msg_arb_session +
		//			" m_FirstMsgSeqNum=" + m_FirstMsgSeqNum +
		//			" msg_arb_seq=" + msg_arb_seq +
		//			" isRecoveryOver returns false");    		    		
		return false;
	}

    
    
    
	public class PeriodicRecovery implements Runnable
	{
	    public PeriodicRecovery() { }


	    public void run()
	    {
	            try
	            {
	            	requestDBRecoveryBatch();
	            }
	            catch (Exception ex)
	            {
	            	s_logger.error(new MessageText("SubscriptionPaginator exception running requestDBRecoveryBatch", ex));	            
	            }
	    }
	 }
    
    
    
}
